package com.example.wordcount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * Created with IntelliJ IDEA.
 * ClassName: WordCountFlink
 * Package: com.example.wordcount
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-07-17
 * Time: 10:26
 */

/**
 * 这种代码实现是DateSetAPI 也就是对数据的处理转换 是看作数据集才操作的
 * 说白了就是 批处理 但是Flink是流处理框架 批量的数据集本质也是流
 * 所以直接使用DataStreamAPI
 */

public class WordCountDataSet {
    public static void main(String[] args) throws Exception {

        //1.创建执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        //2.读取数据 从文件获取
        //相对路径 默认情况下 相对的是工程根路径
        DataSource<String> lineDS = env.readTextFile("input/word.txt");

        //3.切分 转换(word,1)
        FlatMapOperator<String, Tuple2<String, Integer>> wordCount = lineDS.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                //1.按照空格切分单词
                String[] words = s.split(" ");
                //将单词转换为(word,1)
                for (String word : words) {
                    Tuple2<String, Integer> of = Tuple2.of(word, 1);
                    //使用采集器 Collector 向下游发送
                    collector.collect(of);
                }
            }
        });

        //(hello,1) 二元组 的索引
        //4.按照word分组
        UnsortedGrouping<Tuple2<String, Integer>> tuple2UnsortedGrouping = wordCount.groupBy(0);

        //5.各分组聚合 (hello,1) 二元组 的索引
        AggregateOperator<Tuple2<String, Integer>> sum = tuple2UnsortedGrouping.sum(1);


        //6.输出
        sum.print();


    }
}
